package com.cnebula.dataprocess.sqoop.job.service.impl;

import java.util.List;
import java.util.Map;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.apache.sqoop.client.SqoopClient;
import org.apache.sqoop.client.SubmissionCallback;
import org.apache.sqoop.model.MDriverConfig;
import org.apache.sqoop.model.MJob;
import org.apache.sqoop.model.MStringInput;
import org.apache.sqoop.model.MSubmission;
import org.apache.sqoop.validation.Message;
import org.apache.sqoop.validation.Status;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.cnebula.dataprocess.common.JobStatus;
import com.cnebula.dataprocess.common.sqoop.SqoopJobResult;
import com.cnebula.dataprocess.sqoop.job.HDFSJobConfigAdapter;
import com.cnebula.dataprocess.sqoop.job.JDBCJobConfigAdapter;
import com.cnebula.dataprocess.sqoop.job.LinkConstants;
import com.cnebula.dataprocess.sqoop.job.adapter.MLinkHDFSAdapter;
import com.cnebula.dataprocess.sqoop.job.adapter.MLinkJDBCAdapter;
import com.cnebula.dataprocess.sqoop.job.adapter.SqoopJobConfigAdapter;
import com.cnebula.dataprocess.sqoop.job.adapter.SqoopMLinkAdapter;
import com.cnebula.dataprocess.sqoop.job.config.SqoopConfig;
import com.cnebula.dataprocess.sqoop.job.exception.NotFoundFromJobAdapter;
import com.cnebula.dataprocess.sqoop.job.exception.NotFoundLinkAdaper;
import com.cnebula.dataprocess.sqoop.job.exception.NotFoundToJobAdapter;
import com.cnebula.dataprocess.sqoop.job.exception.SqoopJobCreateException;
import com.cnebula.dataprocess.sqoop.job.exception.SqoopMLinkCreateException;
import com.cnebula.dataprocess.sqoop.job.model.LinkConfig;
import com.cnebula.dataprocess.sqoop.job.model.LspMLink;
import com.cnebula.dataprocess.sqoop.job.model.ToSqoopJobConstants;
import com.cnebula.dataprocess.sqoop.job.producer.SqoopJobProducer;
import com.cnebula.dataprocess.sqoop.job.service.SqoopService;

@Component
public class SqoopServiceImpl implements SqoopService {

	SqoopClient client;

	@Resource
	SqoopConfig sqoopConf;

	@Autowired
	SqoopJobProducer producer;

	private static final Logger log = LoggerFactory.getLogger(SqoopServiceImpl.class);

	@PostConstruct
	public void init() {
		client = new SqoopClient(sqoopConf.getUrl());
	}

	@Override
	public LspMLink createMLink(LinkConfig config) throws NotFoundLinkAdaper, SqoopMLinkCreateException {
		String type = config.getType();
		LspMLink link = null;
		SqoopMLinkAdapter adapter = null;
		if (type.equals(LinkConstants.TYPE_MYSQL)) {
			adapter = MLinkJDBCAdapter.getInstance();
		} else if (type.equals(LinkConstants.TYPE_HDFS)) {
			adapter = MLinkHDFSAdapter.getInstance();
		}
		if (adapter == null) {
			throw new NotFoundLinkAdaper(type);
		}
		link = adapter.createMLink(client, config);

		return link;
	}

	public MJob createJob(String jobId, LspMLink fromLink, LspMLink toLink, Map<String, String> fromJobConfig,
			Map<String, String> toJobConfig) throws NotFoundFromJobAdapter, NotFoundToJobAdapter {

		String fromLinkType = fromLink.getType();
		String toLinkType = toLink.getType();

		MJob job = client.createJob(fromLink.getName(), toLink.getName());
		job.setName("lsp-job" + jobId);
		job.setCreationUser("lsp");

		SqoopJobConfigAdapter fromJobConfigAdapter = null;
		if (fromLinkType.equals(LinkConstants.TYPE_MYSQL)) {
			fromJobConfigAdapter = JDBCJobConfigAdapter.getInstance();
		} else if (fromLinkType.equals(LinkConstants.TYPE_HDFS)) {
			fromJobConfigAdapter = HDFSJobConfigAdapter.getInstance();
		}

		if (fromJobConfigAdapter == null) {
			throw new NotFoundFromJobAdapter(fromLinkType);
		}

		SqoopJobConfigAdapter toJobConfigAdapter = null;
		if (toLinkType.equals(LinkConstants.TYPE_MYSQL)) {
			toJobConfigAdapter = JDBCJobConfigAdapter.getInstance();
		} else if (toLinkType.equals(LinkConstants.TYPE_HDFS)) {
			toJobConfigAdapter = HDFSJobConfigAdapter.getInstance();
		}

		if (toJobConfigAdapter == null) {
			throw new NotFoundToJobAdapter(toLinkType);
		}

		fromJobConfigAdapter.setMFromConfig(job, fromJobConfig);
		toJobConfigAdapter.setMToConfig(job, toJobConfig);
		MDriverConfig driverConfig = job.getDriverConfig();
		driverConfig.getIntegerInput("throttlingConfig.numExtractors").setValue(10);

		client.saveJob(job);

		return job;
	}

	@Override
	public void submitJob(String taskId, String jobId, LspMLink fromLink, LspMLink toLink, Map<String, String> fromJobConfig,
			Map<String, String> toJobConfig) throws Exception {

		MJob job = this.createJob(jobId, fromLink, toLink, fromJobConfig, toJobConfig);
		Status status = job.getValidationStatus();
		if (!status.canProceed()) {
			List<Message> validationMessages = job.getValidationMessages();
			String failMsg = getFailMessage(validationMessages);
			throw new SqoopJobCreateException(failMsg);
		}

		try {
			client.startJob(job.getName(), new SubmissionCallback() {

				@Override
				public void updated(MSubmission submission) {
					log.debug("sqoop job:[" + job.getName() + "] updated");
				}

				@Override
				public void submitted(MSubmission submission) {
					log.info("sqoop job:[" + job.getName() + "] submitted");
				}

				@Override
				public void finished(MSubmission submission) {
					log.info("sqoop job:[" + job.getName() + "] finished");
					
					MStringInput input = job.getToJobConfig().getStringInput(ToSqoopJobConstants.HDFS_OUTPUTDIRECTORY);
					String datafile = input.getValue();
					
					SqoopJobResult body = new SqoopJobResult();
					body.setTaskId(taskId);
					body.setJobId(jobId);
					body.setDatafile(datafile);

					producer.sendMsg(body);
				}
			}, 1000 * 10);

		} catch (Exception e) {
			log.error(e.getLocalizedMessage(), e);

			SqoopJobResult body = new SqoopJobResult();
			body.setTaskId(taskId);
			body.setJobId(jobId);
			body.setStatus(JobStatus.FAIL);
			body.setMessage(e.getLocalizedMessage());

			producer.sendMsg(body);
		}

	}

	private String getFailMessage(List<Message> validationMessages) {
		String detailMsg = "";
		for (Message msg : validationMessages) {
			if (!msg.getStatus().canProceed()) {
				detailMsg = msg.getMessage();
				break;
			}
		}

		return detailMsg;
	}

}
